package com.cctc.rds.raft.sdk.client;

import com.cctc.rds.nrpc.api.client.config.RpcClientConfig;
import com.cctc.rds.nrpc.core.client.NettyRpcClient;
import com.cctc.rds.raft.core.RaftClusterConfig;
import com.cctc.rds.raft.core.node.NodeIpPort;

import java.io.Closeable;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

public class RaftClient implements Closeable {
    private final int nodeId;
    private volatile Map<Integer, NettyRpcClient> clients = new HashMap<>();

    public RaftClient(int nodeId) {
        this.nodeId = nodeId;
    }

    public int getNodeId() {
        return nodeId;
    }

    public NettyRpcClient getNettyRpcClient(int nodeId) {
        return clients.get(nodeId);
    }

    public synchronized void connect(Set<NodeIpPort> nodes, RaftClusterConfig config) {
        Map<Integer, NettyRpcClient> clients = new HashMap<>();
        for (NodeIpPort toNode : nodes) {
            if (toNode.getNodeId() == this.nodeId) continue;
            NettyRpcClient client = new NettyRpcClient();
            RpcClientConfig clientConfig = RpcClientConfig.builder()
                    .remoteIp(toNode.getIp())
                    .remotePort(toNode.getPort())
                    .connectionTimeoutMs(config.getHeartBeatMs())
                    .waitResponseTimeoutMs(config.getHeartBeatMs())
                    .failRetryConnectMs(config.getHeartBeatMs())
                    .idleStateMs(config.getHeartBeatMs())
                    .build();
            client.start(clientConfig);
            clients.put(toNode.getNodeId(), client);
        }
        this.clients = clients;
    }

    @Override
    public void close() throws IOException {
        if (clients != null && clients.size() > 0) {
            clients.values().forEach(client -> {
                try {
                    client.stop();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
    }
}
